package com.xxx.flink.transform;

import com.xxx.flink.customsource.CustomSource;
import com.xxx.flink.pojo.Event;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 转换算子FlatMap
 * @author guwq
 * @since 2023/7/26
 */
public class TransformFlatMapTest {

    public static void main(String[] args) throws Exception {
        // 流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据
        DataStreamSource<Event> sourceStream = env.addSource(new CustomSource());

        // FlatMap方式1：静态内部类
//        sourceStream.flatMap(new MyFlatMap()).print();

        // FlatMap方式2：匿名内部类
//        sourceStream.flatMap(new FlatMapFunction<Event, String>() {
//            @Override
//            public void flatMap(Event event, Collector<String> collector) throws Exception {
//                collector.collect(event.getName());
//                collector.collect(event.getUrl());
//            }
//        }).print();

        // FlatMap方式3：Lambda表达式（匿名内部类里只有一个方法）
        sourceStream.flatMap((Event event, Collector<String> collector) -> {
            collector.collect(event.getName());
            collector.collect(event.getUrl());
        }).returns(String.class).print();   // Lambda表达式需要使用returns()方法指明返回类型

        // 提交执行
        env.execute();
    }

    public static class MyFlatMap implements FlatMapFunction<Event, String> {

        @Override
        public void flatMap(Event event, Collector<String> collector) throws Exception {
            collector.collect(event.getName());
            collector.collect(event.getUrl());
        }
    }

}
